package com.lzq.mq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.constants.NameServerConstants;
import com.lzq.mq.common.dto.HeartBeatDTO;
import com.lzq.mq.common.dto.ServiceRegistryRespDTO;
import com.lzq.mq.common.enums.NameServerResponseCode;
import com.lzq.mq.common.event.Listener;
import com.lzq.mq.nameserver.common.CommonCache;
import com.lzq.mq.nameserver.enums.ReplicationMsgTypeEnum;
import com.lzq.mq.nameserver.event.model.HeartBeatEvent;
import com.lzq.mq.nameserver.event.model.ReplicationMsgEvent;
import com.lzq.mq.nameserver.store.ServiceInstance;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.util.UUID;

@Slf4j
public class HeartBeatListener implements Listener<HeartBeatEvent> {
    @Override
    public void onReceive(HeartBeatEvent event) throws IllegalAccessException {

        ChannelHandlerContext channelHandlerContext = event.getChannelHandlerContext();

        Attribute<String> reqId =
                channelHandlerContext.channel()
                        .attr(AttributeKey.valueOf(NameServerConstants.NAMESERVER_CHANNEL_ID));
        String address = reqId.get();
        // 没有做过认证
        if (address == null) {
            ServiceRegistryRespDTO serviceRegistryRespDTO = new ServiceRegistryRespDTO();
            serviceRegistryRespDTO.setMsgId(event.getMsgId());

            TcpMsg tcpMsg = new TcpMsg(
                    NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode(),
                    JSON.toJSONBytes(serviceRegistryRespDTO)
            );
            channelHandlerContext.writeAndFlush(tcpMsg);
            channelHandlerContext.close();
            throw new IllegalAccessException("收到未验证的服务心跳！");
        }

        // 更新服务实例的上次心跳时间
        // 更新服务实例的上次心跳时间
        String[] reqInfoStrArr = address.split(":");
        long currentTimestamp = System.currentTimeMillis();

        ServiceInstance serviceInstance = new ServiceInstance();
        serviceInstance.setIp(reqInfoStrArr[0]);
        serviceInstance.setPort(Integer.valueOf(reqInfoStrArr[1]));
        serviceInstance.setLastHeartBeatTime(currentTimestamp);

        HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
        heartBeatDTO.setMsgId(event.getMsgId());

        TcpMsg tcpMsg = new TcpMsg(
                NameServerResponseCode.HEART_BEAT_SUCCESS.getCode(),
                JSON.toJSONBytes(heartBeatDTO)
        );
        channelHandlerContext.writeAndFlush(tcpMsg);

        CommonCache.getServiceInstanceManager().putIfExist(serviceInstance);    // 预防错误删除

        // 复制到下游节点
        ReplicationMsgEvent replicationMsgEvent = new ReplicationMsgEvent();
        replicationMsgEvent.setServiceInstance(serviceInstance);
        replicationMsgEvent.setMsgId(UUID.randomUUID().toString());
        replicationMsgEvent.setChannelHandlerContext(event.getChannelHandlerContext());
        replicationMsgEvent.setType(ReplicationMsgTypeEnum.HEART_BEAT.getCode());
        CommonCache.getReplicationMsgQueueManager().put(replicationMsgEvent);
    }
}
